package com.fanxuankai.boot.mqbroker.consumer.autoconfigure;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.TypeReference;
import com.fanxuankai.boot.mqbroker.Event;
import com.fanxuankai.boot.mqbroker.autoconfigure.MqBrokerProperties;
import com.fanxuankai.boot.mqbroker.consumer.AbstractMqConsumer;
import com.fanxuankai.boot.mqbroker.consumer.EventListenerRegistry;
import com.fanxuankai.boot.mqbroker.consumer.MqConsumer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.core.AcknowledgeMode;
import org.springframework.amqp.core.AmqpAdmin;
import org.springframework.amqp.core.Queue;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.listener.MessageListenerContainer;
import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;
import org.springframework.amqp.rabbit.listener.api.ChannelAwareMessageListener;
import org.springframework.boot.autoconfigure.condition.ConditionalOnClass;
import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.context.annotation.Bean;

/**
 * @author fanxuankai
 */
@ConditionalOnClass({AmqpAdmin.class})
@ConditionalOnProperty(name = MqBrokerConsumerProperties.ENABLED, matchIfMissing = true)
public class MqBrokerConsumerRabbitAutoConfiguration {
    private static final Logger LOGGER = LoggerFactory.getLogger(MqBrokerConsumerRabbitAutoConfiguration.class);

    @Bean
    public MessageListenerContainer simpleMessageQueueLister(ConnectionFactory connectionFactory,
                                                             AbstractMqConsumer<Event<String>> mqConsumer,
                                                             AmqpAdmin amqpAdmin,
                                                             MqBrokerConsumerProperties properties,
                                                             EventListenerRegistry eventListenerRegistry) {
        SimpleMessageListenerContainer container = new SimpleMessageListenerContainer(connectionFactory);
        // 监听的队列
        container.setQueues(eventListenerRegistry.getTopics().stream().map(Queue::new)
                .peek(amqpAdmin::declareQueue).toArray(Queue[]::new));
        // 是否重回队列
        container.setDefaultRequeueRejected(false);
        container.setErrorHandler(throwable -> LOGGER.error("消费异常", throwable));
        if (properties.isManualAcknowledge()) {
            // 手动确认
            container.setAcknowledgeMode(AcknowledgeMode.MANUAL);
            container.setMessageListener((ChannelAwareMessageListener) (message, channel) -> {
                String json = new String(message.getBody());
                Event<String> event = JSON.parseObject(json, new TypeReference<Event<String>>() {
                });
                mqConsumer.accept(event);
                channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
            });
        } else {
            // 自动确认
            container.setAcknowledgeMode(AcknowledgeMode.AUTO);
            container.setMessageListener(message -> {
                String json = new String(message.getBody());
                Event<String> event = JSON.parseObject(json, new TypeReference<Event<String>>() {
                });
                mqConsumer.accept(event);
            });
        }
        return container;
    }

    @Bean
    @ConditionalOnMissingBean(MqConsumer.class)
    public AbstractMqConsumer<Event<String>> mqConsumer(MqBrokerProperties properties) {
        return new AbstractMqConsumer<Event<String>>(properties) {
        };
    }
}
